"""
Filesystem 模块 - 本地文件系统实现
"""
import os
import shutil
from pathlib import Path
from typing import AsyncIterator, List

import aiofiles
import aiofiles.os
import fsspec
import asyncio

from axiom_boot.conf.manager import Settings
from axiom_boot.di import autowired, service, primary, conditional_on_setting
from axiom_boot.filesystem.exceptions import FileSystemException, FileNotFoundException
from axiom_boot.filesystem.interfaces import IFileSystem
from axiom_boot.logging.setup import get_logger

logger = get_logger(__name__)


@service(name="localfilesystem")
@primary
@conditional_on_setting(key="filesystem__enabled")
@conditional_on_setting(key="filesystem__provider", expected_value="local")
class LocalFileSystem(IFileSystem):
    """
    一个使用本地文件系统作为后端的 IFileSystem 实现。
    它将“桶”映射为根存储路径下的子目录。
    """

    def __init__(self, settings: Settings = autowired()):
        self._settings = settings
        self._root_path_str = str(Path(self._settings.filesystem.local_root_path).resolve())
        self._fs: fsspec.AbstractFileSystem = fsspec.filesystem("file", asynchronous=True)
        
        # 确保根目录存在
        self._fs.makedirs(self._root_path_str, exist_ok=True)
        logger.info(f"本地文件系统已初始化，根目录: '{self._root_path_str}'。")

    def _get_bucket_path(self, bucket_name: str) -> str:
        """安全地获取桶（目录）的绝对路径。"""
        # 移除非法字符，防止路径遍历
        safe_bucket_name = "".join(c for c in bucket_name if c.isalnum() or c in ('-', '_'))
        if not safe_bucket_name:
            raise FileSystemException(f"非法的桶名称: '{bucket_name}'")
        return os.path.join(self._root_path_str, safe_bucket_name)

    def _get_full_path(self, path: str, bucket_name: str | None = None) -> str:
        """安全地获取文件的绝对路径。"""
        if bucket_name is None:
            raise FileSystemException("使用本地文件系统时，必须明确指定 bucket_name。")
        
        bucket_path = self._get_bucket_path(bucket_name)
        full_path = os.path.join(bucket_path, path)
        
        # 再次确认路径安全性
        if not str(Path(full_path).resolve()).startswith(str(Path(bucket_path).resolve())):
            raise FileSystemException(f"检测到非法的路径访问: '{path}'")
        return full_path

    # --- 桶/目录级别操作 ---

    async def create_bucket(self, bucket_name: str, **kwargs) -> bool:
        bucket_path = self._get_bucket_path(bucket_name)
        self._fs.makedirs(bucket_path, exist_ok=True)
        logger.debug(f"本地目录 (桶) '{bucket_name}' 已确保存在于: {bucket_path}")
        return True

    async def delete_bucket(self, bucket_name: str, **kwargs) -> bool:
        bucket_path = self._get_bucket_path(bucket_name)
        if not await self.bucket_exists(bucket_name):
            logger.warning(f"尝试删除不存在的本地目录 (桶): '{bucket_name}'")
            return False
        try:
            # [修复] 使用 asyncio.to_thread 来异步执行会阻塞的同步 I/O 操作
            await asyncio.to_thread(shutil.rmtree, bucket_path)
            logger.info(f"本地目录 (桶) '{bucket_name}' 已被删除。")
            return True
        except Exception as e:
            logger.error(f"删除本地目录 (桶) '{bucket_name}' 时失败: {e}", exc_info=True)
            raise FileSystemException(f"无法删除目录 '{bucket_name}'")

    async def bucket_exists(self, bucket_name: str, **kwargs) -> bool:
        bucket_path = self._get_bucket_path(bucket_name)
        return await aiofiles.os.path.isdir(bucket_path)

    async def list_buckets(self) -> List[str]:
        # fsspec.ls 返回的是包含路径信息的字典列表
        sub_dirs = await self._fs._ls(self._root_path_str, detail=True)
        return [
            Path(d['name']).name for d in sub_dirs if d['type'] == 'directory'
        ]

    # --- 文件/对象级别操作 ---

    async def save(self, stream: AsyncIterator[bytes], path: str, bucket_name: str | None = None, **kwargs) -> str:
        full_path = self._get_full_path(path, bucket_name)
        
        parent_dir = os.path.dirname(full_path)
        # 使用 fsspec 的同步方法创建目录
        self._fs.makedirs(parent_dir, exist_ok=True)

        try:
            # 使用 aiofiles 进行真正的异步文件写入
            async with aiofiles.open(full_path, 'wb') as f:
                async for chunk in stream:
                    await f.write(chunk)
            logger.debug(f"文件已成功保存到: {full_path}")
            return full_path
        except Exception as e:
            logger.error(f"保存文件到 '{full_path}' 时发生错误: {e}", exc_info=True)
            raise FileSystemException(f"无法保存文件 '{path}'")

    async def read(self, path: str, bucket_name: str | None = None, chunk_size: int = 8192, **kwargs) -> AsyncIterator[bytes]:
        full_path = self._get_full_path(path, bucket_name)
        
        if not await self.exists(path, bucket_name):
            raise FileNotFoundException(f"文件不存在: '{path}' in bucket '{bucket_name}'")
            
        try:
            # 使用 aiofiles 进行真正的异步文件读取
            async with aiofiles.open(full_path, 'rb') as f:
                while True:
                    chunk = await f.read(chunk_size)
                    if not chunk:
                        break
                    yield chunk
        except Exception as e:
            logger.error(f"读取文件 '{full_path}' 时发生错误: {e}", exc_info=True)
            raise FileSystemException(f"无法读取文件 '{path}'")

    async def delete(self, path: str, bucket_name: str | None = None, **kwargs) -> bool:
        if not await self.exists(path, bucket_name):
            logger.warning(f"尝试删除一个不存在的文件: {path}")
            return False
            
        full_path = self._get_full_path(path, bucket_name)
        await asyncio.to_thread(self._fs.rm, full_path, recursive=False)
        logger.debug(f"文件已成功删除: {full_path}")
        return True

    async def exists(self, path: str, bucket_name: str | None = None, **kwargs) -> bool:
        full_path = self._get_full_path(path, bucket_name)
        return await asyncio.to_thread(self._fs.exists, full_path)

    async def get_presigned_url(self, path: str, bucket_name: str | None = None, expires_in: int = 3600, **kwargs) -> str:
        full_path = self._get_full_path(path, bucket_name)
        # 本地文件系统没有预签名 URL 的概念，返回 'file://' URI
        return f"file://{full_path}"
